package com.xbai.spark.recall.engine

import com.xbai.spark.recall.engine.training.ItemCFTraining
import org.apache.spark.sql.functions.desc
import org.apache.spark.sql.{DataFrame, SparkSession, functions}

/**
  * @author xbai
  * @Date 2021/1/23
  */
class ItemCFEngine(spark: SparkSession) {

  var itemSimilarCache: DataFrame = _
  var itemJaccardSimilarCache: DataFrame = _
  var userItemRatingCache: DataFrame = _

  def init(): Unit = {
    userItemRatingCache = spark.sql("select * from userItemRating")
    val itemCFTraining = new ItemCFTraining()
    itemCFTraining.trainingEngine(userItemRatingCache, spark)
    itemSimilarCache = spark.sql("select * from itemSimilar")
    itemCFTraining.trainEngineWithJaccard(userItemRatingCache, spark)
    itemJaccardSimilarCache = spark.sql("select * from itemJaccardSimilar")
  }

  /**
    * 获取物品最相似的topn个物品
    * @param findItemId 物品id
    * @param num  最相似的数量
    * @return 物品列表
    */
  def getSimilarItems(findItemId:String, num: Int): DataFrame = {
    val items: DataFrame = itemSimilarCache.filter(functions.col("itemId").equalTo(findItemId))
      .sort(desc("itemsim"))
      .limit(num)
    items
  }

  /**
    * 获取物品最相似的topn个物品
    * @param findItemId 物品id
    * @param num  最相似的数量
    * @return 物品列表
    */
  def getJaccardSimilarItems(findItemId:String, num: Int): DataFrame = {
    val items: DataFrame = itemJaccardSimilarCache.filter(functions.col("itemId").equalTo(findItemId))
      .sort(desc("itemsim"))
      .limit(10)
    items
  }
}
